package com.bs.rocketmq.controller;

import com.alibaba.fastjson.JSON;
import com.bs.common.utils.SnowFlakeUtil;
import com.bs.rocketmq.vo.UserVo;
import lombok.extern.log4j.Log4j2;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.*;

import javax.annotation.Resource;
import java.util.Arrays;
import java.util.Date;
import java.util.List;

/**
 * @author bingshao
 * @date 2022/11/17
 **/
@RestController
@RequestMapping("/produce")
@Log4j2
public class ProducerController {

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @PostMapping("/sendSyncMessage")
    public void sendSyncMessage() {
        UserVo userVo = new UserVo();
        userVo.setId(SnowFlakeUtil.getId());
        userVo.setName("同步消息");
        userVo.setCreateTime(new Date());
        rocketMQTemplate.syncSend("sync-topic", JSON.toJSONString(userVo));
    }

    @PostMapping("/sendDelayMessage")
    public void sendDelayMessage() {
        UserVo userVo = new UserVo();
        userVo.setId(SnowFlakeUtil.getId());
        userVo.setName("延迟消息");
        userVo.setCreateTime(new Date());
        Message<UserVo> message = MessageBuilder.withPayload(userVo).build();
        //messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
        // delay = 2,第二个等级，即延迟5秒
        System.out.println("发送时间" + new Date());
        rocketMQTemplate.syncSend("delay-topic", message, 3000, 2);
    }

    @PostMapping("/sendOneWayMessage")
    public void sendOneWayMessage() {
        UserVo userVo = new UserVo();
        userVo.setId(SnowFlakeUtil.getId());
        userVo.setName("单向消息");
        userVo.setCreateTime(new Date());
        rocketMQTemplate.sendOneWay("oneWay-topic", JSON.toJSONString(userVo));
    }

    @PostMapping("/sendOrderlyMessage")
    public void sendOrderlyMessage() {
        Long id1 = SnowFlakeUtil.getId();
        Long id2 = SnowFlakeUtil.getId();
        List<UserVo> userVoList = Arrays.asList(
                new UserVo(id1, "顺序消息1-1", new Date()),
                new UserVo(id1, "顺序消息1-2", new Date()),
                new UserVo(id1, "顺序消息1-3", new Date()),
                new UserVo(id2, "顺序消息2-1", new Date()),
                new UserVo(id2, "顺序消息2-2", new Date()),
                new UserVo(id2, "顺序消息2-3", new Date())
        );
        userVoList.forEach(userVo -> rocketMQTemplate.syncSendOrderly("orderly-topic", JSON.toJSONString(userVo), userVo.getId().toString()));

    }

    @PostMapping("/sendAsyncMessage")
    public void sendAsyncMessage() {
        UserVo userVo = new UserVo();
        Long id = SnowFlakeUtil.getId();
        userVo.setId(id);
        userVo.setName("异步消息");
        userVo.setCreateTime(new Date());

        // 将消息存入数据库或者redis然后再发送
        redisTemplate.opsForValue().set(id.toString(), JSON.toJSONString(userVo));

        rocketMQTemplate.asyncSend("async-topic", JSON.toJSONString(userVo), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("messageQueue===>" + sendResult.getMessageQueue()
                        + ",sendStatus===>" + sendResult.getSendStatus()
                        + ",msgId===>" + sendResult.getMsgId()
                        + ",transactionId===>" + sendResult.getTransactionId()
                        + ",offsetMsgId===>" + sendResult.getOffsetMsgId()
                        + ",queueOffset===>" + sendResult.getQueueOffset()
                        + ",regionId===>" + sendResult.getRegionId());

                redisTemplate.delete(id.toString());
            }

            @Override
            public void onException(Throwable throwable) {
                log.info("exception===>{}", throwable.getMessage());
            }
        });
    }

    @PostMapping("/sendHasKeyMessage")
    public void sendHasKeyMessage() {
        UserVo userVo = new UserVo();
        Long id = SnowFlakeUtil.getId();
        userVo.setId(id);
        userVo.setName("带有key的消息");
        userVo.setCreateTime(new Date());
        Message<UserVo> message = MessageBuilder.withPayload(userVo).setHeader(RocketMQHeaders.KEYS, id.toString()).build();
        rocketMQTemplate.send("hasKey-topic", message);
    }

}
